Crewai

CrewAI는 LangChain이나 다른 에이전트 프레임워크와 완전히 독립적으로 처음부터 구축된 경량화되고 초고속 Python 프레임워크입니다. CrewAI는 개발자에게 높은 수준의 단순성(Crews)과 정밀한 저수준 제어(Flows)를 제공하여 모든 시나리오에 맞는 자율 AI 에이전트를 만드는 데 이상적입니다. CrewAI here.

AI 에이전트로 작업할 때 상호작용을 디버깅하고 모니터링하는 것이 중요합니다. CrewAI 애플리케이션은 종종 여러 에이전트가 함께 작동하므로 이들이 어떻게 협업하고 소통하는지 이해하는 것이 필수적입니다. Weave는 CrewAI 애플리케이션의 추적을 자동으로 캡처하여 에이전트의 성능과 상호작용을 모니터링하고 분석할 수 있게 함으로써 이 과정을 단순화합니다.

이 통합은 Crews와 Flows를 모두 지원합니다.

Crew로 시작하기

이 예제를 실행하려면 CrewAI(more details)와 weave를 설치해야 합니다:

pip install crewai weave

이제 CrewAI Crew를 생성하고 Weave를 사용하여 실행을 추적해 보겠습니다. 시작하려면 스크립트 시작 부분에서 간단히 weave.init()를 호출하세요. weave.init()의 인수는 추적이 기록될 프로젝트 이름입니다.

import weave
from crewai import Agent, Task, Crew, LLM, Process

# Initialize Weave with your project name
# highlight-next-line
weave.init(project_name="crewai_demo")

# Create an LLM with a temperature of 0 to ensure deterministic outputs
llm = LLM(model="gpt-4o-mini", temperature=0)

# Create agents
researcher = Agent(
    role='Research Analyst',
    goal='Find and analyze the best investment opportunities',
    backstory='Expert in financial analysis and market research',
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role='Report Writer',
    goal='Write clear and concise investment reports',
    backstory='Experienced in creating detailed financial reports',
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

# Create tasks
research_task = Task(
    description='Deep research on the {topic}',
    expected_output='Comprehensive market data including key players, market size, and growth trends.',
    agent=researcher
)

writing_task = Task(
    description='Write a detailed report based on the research',
    expected_output='The report should be easy to read and understand. Use bullet points where applicable.',
    agent=writer
)

# Create a crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=True,
    process=Process.sequential,
)

# Run the crew
result = crew.kickoff(inputs={"topic": "AI in material science"})
print(result)

Weave는 에이전트 상호작용, 작업 실행, LLM 호출을 포함하여 CrewAI 라이브러리를 통해 이루어진 모든 호출을 추적하고 기록합니다. Weave 웹 인터페이스에서 추적을 볼 수 있습니다.

CrewAI는 시작 프로세스를 더 잘 제어할 수 있는 여러 메서드를 제공합니다: kickoff(), kickoff_for_each(), kickoff_async(), 그리고 kickoff_for_each_async(). 이 통합은 이러한 모든 메서드에서 로그 추적을 지원합니다.

도구 추적

CrewAI 도구는 에이전트에게 웹 검색, 데이터 분석부터 협업 및 동료 간 작업 위임에 이르기까지 다양한 기능을 제공합니다. 이 통합은 이러한 도구도 추적할 수 있습니다.

인터넷을 검색하고 가장 관련성 높은 결과를 반환할 수 있는 도구에 접근 권한을 부여하여 위 예제에서 생성된 보고서의 품질을 향상시키겠습니다.

먼저 추가 종속성을 설치해 보겠습니다.

pip install 'crewai[tools]'

이 예제에서는 SerperDevTool를 사용하여 ‘Research Analyst’ 에이전트가 인터넷에서 관련 정보를 검색할 수 있도록 합니다. 이 도구와 API 요구 사항에 대해 자세히 알아보려면 here를 참조하세요.

# .... existing imports ....
from crewai_tools import SerperDevTool

# We provide the agent with the tool.
researcher = Agent(
    role='Research Analyst',
    goal='Find and analyze the best investment opportunities',
    backstory='Expert in financial analysis and market research',
    llm=llm,
    verbose=True,
    allow_delegation=False,
    # highlight-next-line
    tools=[SerperDevTool()],
)

# .... existing code ....

인터넷에 접근할 수 있는 에이전트가 있는 이 Crew를 실행하면 더 나은 관련성 높은 결과를 얻을 수 있습니다. 아래 이미지와 같이 도구 사용을 자동으로 추적합니다.

이 통합은 crewAI-tools 저장소에서 사용 가능한 모든 도구를 자동으로 패치합니다.

Flow로 시작하기

import weave
# Initialize Weave with your project name
# highlight-next-line
weave.init("crewai_demo")

from crewai.flow.flow import Flow, listen, router, start
from litellm import completion


class CustomerFeedbackFlow(Flow):
    model = "gpt-4o-mini"

    @start()
    def fetch_feedback(self):
        print("Fetching customer feedback")
        # In a real-world scenario, this could be replaced by an API call.
        # For this example, we simulate customer feedback.
        feedback = (
            "I had a terrible experience with the product. "
            "It broke after one use and customer service was unhelpful."
        )
        self.state["feedback"] = feedback
        return feedback

    @router(fetch_feedback)
    def analyze_feedback(self, feedback):
        # Use the language model to analyze sentiment
        prompt = (
            f"Analyze the sentiment of this customer feedback and "
            "return only 'positive' or 'negative':\n\n"
            f"Feedback: {feedback}"
        )
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        sentiment = response["choices"][0]["message"]["content"].strip().lower()
        # If the response is ambiguous, default to negative
        if sentiment not in ["positive", "negative"]:
            sentiment = "negative"
        return sentiment

    @listen("positive")
    def handle_positive_feedback(self):
        # Generate a thank you message for positive feedback
        prompt = "Generate a thank you message for a customer who provided positive feedback."
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        thank_you_message = response["choices"][0]["message"]["content"].strip()
        self.state["response"] = thank_you_message
        return thank_you_message

    @listen("negative")
    def handle_negative_feedback(self):
        # Generate an apology message with a promise to improve service for negative feedback
        prompt = (
            "Generate an apology message to a customer who provided negative feedback and offer assistance or a solution."
        )
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        apology_message = response["choices"][0]["message"]["content"].strip()
        self.state["response"] = apology_message
        return apology_message

# Instantiate and kickoff the flow
flow = CustomerFeedbackFlow()
result = flow.kickoff()
print(result)

이 통합은 Flow.kickoff 진입점과 사용 가능한 모든 데코레이터를 자동으로 패치합니다 — @start, @listen, @router, @or_ 및 @and_.

Crew 가드레일 - 자체 작업 추적

작업 가드레일은 작업 출력을 다음 작업으로 전달하기 전에 검증하고 변환하는 방법을 제공합니다. 간단한 Python 함수를 사용하여 에이전트의 실행을 실시간으로 검증할 수 있습니다.

이 함수를 @weave.op로 래핑하면 입력, 출력 및 앱 로직을 캡처하여 데이터가 에이전트를 통해 어떻게 검증되는지 디버깅할 수 있습니다. 또한 실험할 때 코드를 자동으로 버전 관리하여 git에 커밋되지 않은 임시 세부 정보를 캡처합니다.

연구 분석가와 작가의 예를 살펴보겠습니다. 생성된 보고서의 길이를 검증하는 가드레일을 추가합니다.

# .... existing imports and weave initialization ....

# Decorate your guardrail function with `@weave.op()`
# highlight-next-line
@weave.op(name="guardrail-validate_blog_content")
def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
    # Get raw string result
    result = result.raw

    """Validate blog content meets requirements."""
    try:
        # Check word count
        word_count = len(result.split())

        if word_count > 200:
            return (False, {
                "error": "Blog content exceeds 200 words",
                "code": "WORD_COUNT_ERROR",
                "context": {"word_count": word_count}
            })

        # Additional validation logic here
        return (True, result.strip())
    except Exception as e:
        return (False, {
            "error": "Unexpected error during validation",
            "code": "SYSTEM_ERROR"
        })


# .... existing agents and research analyst task ....

writing_task = Task(
    description='Write a detailed report based on the research under 200 words',
    expected_output='The report should be easy to read and understand. Use bullet points where applicable.',
    agent=writer,
    # highlight-next-line
    guardrail=validate_blog_content,
)

# .... existing code to run crew ....

가드레일 함수를 @weave.op로 장식하기만 하면 이 함수에 대한 입력과 출력을 실행 시간, LLM이 내부적으로 사용되는 경우 토큰 정보, 코드 버전 등과 함께 추적할 수 있습니다.

결론

이 통합에 대해 개선해야 할 사항이 있으면 알려주세요. 문제가 발생하면 here에 이슈를 열어주세요.

CrewAI를 사용하여 강력한 다중 에이전트 시스템을 구축하는 방법에 대해 자세히 알아보려면 many examples와 documentation을 참조하세요.